package com.gateway.video.rtsp.codec.client;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.CombinedChannelDuplexHandler;
import io.netty.handler.codec.PrematureChannelClosureException;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.rtsp.RtspDecoder;
import io.netty.handler.codec.rtsp.RtspEncoder;
import io.netty.util.ReferenceCountUtil;

import java.util.ArrayDeque;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicLong;

public class RtspClientCodec extends CombinedChannelDuplexHandler<RtspDecoder, RtspEncoder> implements HttpClientUpgradeHandler.SourceCodec {
    private final Queue<HttpMethod> queue;
    private final boolean parseHttpAfterConnectRequest;
    private boolean done;
    private final AtomicLong requestResponseCounter;
    private final boolean failOnMissingResponse;

    public RtspClientCodec() {
        this(4096, 8192, 8192, false);
    }

    public RtspClientCodec(int maxInitialLineLength, int maxHeaderSize, int maxContentLength) {
        this(maxInitialLineLength, maxHeaderSize, maxContentLength, false);
    }

    public RtspClientCodec(int maxInitialLineLength, int maxHeaderSize, int maxContentLength, boolean failOnMissingResponse) {
        this(maxInitialLineLength, maxHeaderSize, maxContentLength, failOnMissingResponse, true);
    }

    public RtspClientCodec(int maxInitialLineLength, int maxHeaderSize, int maxContentLength, boolean failOnMissingResponse, boolean validateHeaders) {
        this(maxInitialLineLength, maxHeaderSize, maxContentLength, failOnMissingResponse, validateHeaders, false);
    }

    public RtspClientCodec(int maxInitialLineLength, int maxHeaderSize, int maxContentLength, boolean failOnMissingResponse, boolean validateHeaders, boolean parseHttpAfterConnectRequest) {
        this.queue = new ArrayDeque();
        this.requestResponseCounter = new AtomicLong();
        this.init(new Decoder(maxInitialLineLength, maxHeaderSize, maxContentLength, validateHeaders), new Encoder());
        this.failOnMissingResponse = failOnMissingResponse;
        this.parseHttpAfterConnectRequest = parseHttpAfterConnectRequest;
    }

    public RtspClientCodec(int maxInitialLineLength, int maxHeaderSize, int maxChunkSize, boolean failOnMissingResponse, boolean validateHeaders, int initialBufferSize) {
        this(maxInitialLineLength, maxHeaderSize, maxChunkSize, failOnMissingResponse, validateHeaders, initialBufferSize, false);
    }

    public RtspClientCodec(int maxInitialLineLength, int maxHeaderSize, int maxContentLength, boolean failOnMissingResponse, boolean validateHeaders, int initialBufferSize, boolean parseHttpAfterConnectRequest) {
        this.queue = new ArrayDeque();
        this.requestResponseCounter = new AtomicLong();
        this.init(new RtspClientCodec.Decoder(maxInitialLineLength, maxHeaderSize, maxContentLength, validateHeaders), new RtspClientCodec.Encoder());
        this.parseHttpAfterConnectRequest = parseHttpAfterConnectRequest;
        this.failOnMissingResponse = failOnMissingResponse;
    }

    @Override
    public void prepareUpgradeFrom(ChannelHandlerContext channelHandlerContext) {
        ((RtspClientCodec.Encoder)this.outboundHandler()).upgraded = true;
    }

    @Override
    public void upgradeFrom(ChannelHandlerContext channelHandlerContext) {
        ChannelPipeline p = channelHandlerContext.pipeline();
        p.remove(this);
    }

    public void setSingleDecode(boolean singleDecode) {
        ((RtspDecoder)this.inboundHandler()).setSingleDecode(singleDecode);
    }

    public boolean isSingleDecode() {
        return ((RtspDecoder)this.inboundHandler()).isSingleDecode();
    }

    private final class Decoder extends RtspDecoder {
        Decoder(int maxInitialLineLength, int maxHeaderSize, int maxContentLength, boolean validateHeaders) {
            super(maxInitialLineLength, maxHeaderSize, maxContentLength, validateHeaders);
        }

        protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
            int oldSize;
            if (RtspClientCodec.this.done) {
                oldSize = this.actualReadableBytes();
                if (oldSize == 0) {
                    return;
                }

                out.add(buffer.readBytes(oldSize));
            } else {
                oldSize = out.size();
                super.decode(ctx, buffer, out);
                if (RtspClientCodec.this.failOnMissingResponse) {
                    int size = out.size();

                    for(int i = oldSize; i < size; ++i) {
                        this.decrement(out.get(i));
                    }
                }
            }

        }

        private void decrement(Object msg) {
            if (msg != null) {
                if (msg instanceof LastHttpContent) {
                    RtspClientCodec.this.requestResponseCounter.decrementAndGet();
                }

            }
        }

        protected boolean isContentAlwaysEmpty(HttpMessage msg) {
            int statusCode = ((HttpResponse)msg).status().code();
            if (statusCode != 100 && statusCode != 101) {
                HttpMethod method = (HttpMethod)RtspClientCodec.this.queue.poll();
                if (method != null) {
                    char firstChar = method.name().charAt(0);
                    switch(firstChar) {
                        case 'C':
                            if (statusCode == 200 && HttpMethod.CONNECT.equals(method)) {
                                if (!RtspClientCodec.this.parseHttpAfterConnectRequest) {
                                    RtspClientCodec.this.done = true;
                                    RtspClientCodec.this.queue.clear();
                                }

                                return true;
                            }
                            break;
                        case 'H':
                            if (HttpMethod.HEAD.equals(method)) {
                                return true;
                            }
                    }
                }

                return super.isContentAlwaysEmpty(msg);
            } else {
                return super.isContentAlwaysEmpty(msg);
            }
        }

        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            super.channelInactive(ctx);
            if (RtspClientCodec.this.failOnMissingResponse) {
                long missingResponses = RtspClientCodec.this.requestResponseCounter.get();
                if (missingResponses > 0L) {
                    ctx.fireExceptionCaught(new PrematureChannelClosureException("channel gone inactive with " + missingResponses + " missing response(s)"));
                }
            }

        }
    }

    private final class Encoder extends RtspEncoder {
        boolean upgraded;

        private Encoder() {
        }

        protected void encode(ChannelHandlerContext ctx, Object msg, List<Object> out) throws Exception {
            if (this.upgraded) {
                out.add(ReferenceCountUtil.retain(msg));
            } else {
                if (msg instanceof HttpRequest) {
                    RtspClientCodec.this.queue.offer(((HttpRequest)msg).method());
                }

                super.encode(ctx, msg, out);
                if (RtspClientCodec.this.failOnMissingResponse && !RtspClientCodec.this.done && msg instanceof LastHttpContent) {
                    RtspClientCodec.this.requestResponseCounter.incrementAndGet();
                }

            }
        }
    }
}
